package com.fxkj.cloud.drms.ui.mqtt.handler;

import com.fxkj.cloud.drms.infra.mq.mqtt.MqttConstants;
import com.fxkj.cloud.drms.ui.mqtt.handler.receive.*;
import com.fxkj.cloud.drms.mqtt.MqttHandler;
import com.google.common.util.concurrent.Futures;
import io.netty.buffer.ByteBuf;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Future;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * <p>description: 默认mqtt消息接收处理器</p>
 * @author ww
 * @date 2025/3/14 11:29
 */
@Component
@Slf4j
public class DefaultMqttReceiveHandler implements MqttHandler {

    private final Map<Pattern, AbstractReceiveHandler> handlerMap;

    public DefaultMqttReceiveHandler(ApplicationContext context) {
        this.handlerMap = new HashMap<>();
        //添加消息接收处理器
        handlerMap.put(MqttConstants.Topic.Receive.DEVICE_DATA_UPLOAD_PATTERN, context.getBean(MqttConstants.Topic.Receive.DEVICE_MSG_RECEIVER_SERVICE_NAME, DeviceMsgReceiverHandler.class));
        handlerMap.put(MqttConstants.Topic.System.CLIENT_ONLINE_PATTERN, context.getBean(MqttConstants.Topic.System.SYSTEM_DEVICE_ONLINE_SERVICE_NAME, SystemDeviceOnlineHandler.class));
        handlerMap.put(MqttConstants.Topic.System.CLIENT_OFFLINE_PATTERN, context.getBean(MqttConstants.Topic.System.SYSTEM_DEVICE_OFFLINE_SERVICE_NAME, SystemDeviceOfflineHandler.class));
        handlerMap.put(MqttConstants.Topic.Receive.DEVICE_PLAN_COMMAND_RESPONSE_PATTERN, context.getBean(MqttConstants.Topic.Receive.DEVICE_PLAN_COMMAND_RESPONSE_SERVICE_NAME, DevicePlanCommandResponseHandler.class));
        handlerMap.put(MqttConstants.Topic.Receive.DEVICE_ALARM_CREATE_PATTERN, context.getBean(MqttConstants.Topic.Receive.DEVICE_ALARM_CREATE_SERVICE_NAME, DeviceAlarmCreateReceiverHandler.class));
        handlerMap.put(MqttConstants.Topic.Receive.DEVICE_ALARM_CLOSE_PATTERN, context.getBean(MqttConstants.Topic.Receive.DEVICE_ALARM_CLOSE_SERVICE_NAME, DeviceAlarmCloseReceiverHandler.class));
    }

    @Override
    public Future<Void> onMessage(String topic, ByteBuf payload) {
        log.trace("[DefaultMqttReceiveHandler] -| receive msg from topic: {}", topic);
        handlerMap.entrySet().stream().filter(e -> e.getKey().matcher(topic).matches()).findFirst().ifPresentOrElse(map -> {
            Matcher matcher = null;
            try {
                matcher = map.getKey().matcher(topic);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
            if (matcher.find()) {
                try {
                    map.getValue().doHandler(matcher, topic, payload);
                } catch (Exception e) {
                    log.error("handle msg error", e);
                    throw new RuntimeException(e);
                }
            }
        }, () -> {
            log.info("no handler for topic: {}", topic);
        });
        return Futures.immediateVoidFuture();
    }
}
